package drds.data_migrate.task_group.task.data_exchange.exchanger;


import drds.common.Author;
import drds.data_migrate.Authors;
import drds.data_migrate.task_group.task.data_exchange.record.DefaultRecord;
import drds.data_migrate.task_group.task.data_exchange.record.Record;
import drds.data_migrate.task_group.task.data_exchange.record.TerminateRecord;
import drds.data_migrate.vo.TaskConfig;
import lombok.NonNull;

import java.util.ArrayList;
import java.util.List;

@Author(name = Authors.LI_YANG)
public class RecordExchanger implements Produce, Consume
{


    protected final int byteCapacity;
    //
    private final RecordQueue recordQueue;
    private final List<Record> recordList;
    private final TaskConfig taskConfig;

    //

    private int recordListSize;
    private int recordListIndex = 0;
    private volatile boolean shutdown = false;

    @SuppressWarnings("unchecked")
    public RecordExchanger(@NonNull  final RecordQueue recordQueue)
    {


        this.recordQueue = recordQueue;
        this.taskConfig = recordQueue.getTaskConfig();
        //
        this.recordListSize = taskConfig.getRecordListSize();
        this.recordList = new ArrayList<Record>(recordListSize);

        //channel的queue默认大小为8M，原来为64M
        this.byteCapacity = taskConfig.getTransportChannelCapacityByte();


    }

    /**
     * <pre>
     * ========================================================================================================================
     * Produce
     * ========================================================================================================================
     * </pre>
     */

    public Record createRecord()
    {
        return new DefaultRecord();
    }


    public void addToRecordList(@NonNull Record record)
    {
        if (shutdown)
        {
            throw new RuntimeException();
        }


        boolean isFull = (this.recordListIndex >= this.recordListSize);
        if (isFull)
        {
            pushAllToRecordQueue();
        }

        this.recordList.add(record);
        this.recordListIndex++;

    }


    public void pushAllToRecordQueue()
    {
        if (shutdown)
        {
            throw new RuntimeException();
        }
        this.recordQueue.pushAll(this.recordList);
        this.recordList.clear();
        this.recordListIndex = 0;

    }


    public void terminate()
    {
        if (shutdown)
        {
            throw new RuntimeException();
        }
        pushAllToRecordQueue();
        this.recordQueue.pushTerminateRecord(TerminateRecord.get());
    }

    public void shutdown()
    {
        shutdown = true;
        try
        {
            recordList.clear();
            recordQueue.clear();
        } catch (Throwable t)
        {
            t.printStackTrace();
        }
    }

    /**
     * <pre>
     * ========================================================================================================================
     * Consume
     * ========================================================================================================================
     * </pre>
     */

    public Record getFromRecordList()
    {
        if (shutdown)
        {
            throw new RuntimeException();
        }
        boolean isEmpty = (this.recordListIndex >= this.recordList.size());
        if (isEmpty)
        {
            pullAllFromRecordQueue();
        }

        Record record = this.recordList.get(this.recordListIndex++);
        if (record instanceof TerminateRecord)
        {
            record = null;
        }
        return record;
    }


    private void pullAllFromRecordQueue()
    {
        this.recordQueue.pullAll(this.recordList);
        this.recordListIndex = 0;
        this.recordListSize = this.recordList.size();
    }
}
